/* ==================================================================
 * JdbcMqttMessageDao.java - 11/06/2021 4:10:17 PM
 * 
 * Copyright 2021 SolarNetwork.net Dev Team
 * 
 * This program is free software; you can redistribute it and/or 
 * modify it under the terms of the GNU General Public License as 
 * published by the Free Software Foundation; either version 2 of 
 * the License, or (at your option) any later version.
 * 
 * This program is distributed in the hope that it will be useful, 
 * but WITHOUT ANY WARRANTY; without even the implied warranty of 
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 
 * General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License 
 * along with this program; if not, write to the Free Software 
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 
 * 02111-1307 USA
 * ==================================================================
 */

package net.solarnetwork.node.dao.mqtt.jdbc;

import static java.lang.String.format;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.Instant;
import java.util.Collections;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.atomic.AtomicLong;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import net.solarnetwork.common.mqtt.MqttQos;
import net.solarnetwork.common.mqtt.MqttStats;
import net.solarnetwork.common.mqtt.dao.BasicMqttMessageEntity;
import net.solarnetwork.common.mqtt.dao.MqttMessageDao;
import net.solarnetwork.common.mqtt.dao.MqttMessageEntity;
import net.solarnetwork.node.dao.jdbc.BaseJdbcBatchableDao;
import net.solarnetwork.service.PingTest;
import net.solarnetwork.service.PingTestResult;
import net.solarnetwork.settings.SettingSpecifier;
import net.solarnetwork.settings.SettingSpecifierProvider;
import net.solarnetwork.settings.support.BasicTitleSettingSpecifier;

/**
 * JDBC implementation of {@link MqttMessageDao}.
 * 
 * @author matt
 * @version 2.1
 */
public class JdbcMqttMessageDao extends BaseJdbcBatchableDao<MqttMessageEntity, Long>
		implements MqttMessageDao, SettingSpecifierProvider, PingTest {

	/**
	 * The default SQL template for the {@code sqlGetTablesVersion} property.
	 * The {@link #getTableName()} value is used in the pattern, e.g.
	 * {@code T-init.sql}.
	 */
	public static final String SQL_GET_TABLES_VERSION_TEMPLATE = "SELECT svalue FROM solarnode.mqtt_message_meta WHERE skey = 'solarnode.%s.version'";

	/** The table name for {@link MqttMessageEntity} entities. */
	public static final String TABLE_NAME = "message";

	/** The charge point table version. */
	public static final int VERSION = 1;

	/** The {@code maxCountPingFail} property default value. */
	public static final int DEFAULT_MAX_COUNT_PING_FAIL = 10000;

	/**
	 * Enumeration of SQL resources.
	 */
	public enum SqlResource {

		/** Batch list. */
		Batch("batch"),

		/** Batch list for a specific destination. */
		BatchForDestination("batch-for-destination"),

		/** Batch update. */
		BatchUpdate("batch-update"),

		/** Batch update for a specific destination. */
		BatchUpdateForDestination("batch-update-for-destination"),

		/** Get a count of stored records. */
		Count("count"),

		;

		private final String resource;

		private SqlResource(String resource) {
			this.resource = resource;
		}

		/**
		 * Get the SQL resource name.
		 * 
		 * @return the resource
		 */
		public String getResource() {
			return resource;
		}
	}

	public final MqttStats stats;
	private int maxCountPingFail = DEFAULT_MAX_COUNT_PING_FAIL;

	/**
	 * Constructor.
	 */
	public JdbcMqttMessageDao() {
		super(BasicMqttMessageEntity.class, Long.class, MqttMessageEntityRowMapper.INSTANCE, "mqtt_%s",
				TABLE_NAME, VERSION);
		setSqlGetTablesVersion(format(SQL_GET_TABLES_VERSION_TEMPLATE, getTableName()));
		setUseAutogeneratedKeys(true);
		this.stats = new MqttStats("JdbcMqttMessageDao", 100, MqttMessageDaoStat.values());
	}

	@Override
	public String getSettingUid() {
		return "net.solarnetwork.node.dao.mqtt.jdbc";
	}

	@Override
	public String getDisplayName() {
		return "MqttMessageDao (JDBC)";
	}

	@Override
	public List<SettingSpecifier> getSettingSpecifiers() {
		return Collections.singletonList((SettingSpecifier) new BasicTitleSettingSpecifier("status",
				getStatusMessage(), true, true));
	}

	private String getStatusMessage() {
		// @formatter:off
		long rowCount = 0;
		try {
			rowCount = rowCount();
		} catch ( Exception e ) {
			log.warn("Error finding MQTT message row count.", e);
		}
		return getMessageSource().getMessage("status.msg",
				new Object[] { 
						rowCount,
						stats.get(MqttMessageDaoStat.MessagesStored),
						stats.get(MqttMessageDaoStat.MessagesDeleted) },
				Locale.getDefault());
		// @formatter:on
	}

	private long rowCount() {
		final Number rowCountNum = getJdbcTemplate()
				.queryForObject(getSqlResource(SqlResource.Count.getResource()), Number.class);
		return (rowCountNum == null ? 0 : rowCountNum.longValue());
	}

	@Transactional(readOnly = false, propagation = Propagation.REQUIRED)
	@Override
	public Long save(MqttMessageEntity entity) {
		Long id = super.save(entity);
		stats.incrementAndGet(MqttMessageDaoStat.MessagesStored);
		return id;
	}

	@Transactional(readOnly = false, propagation = Propagation.REQUIRED)
	@Override
	public void delete(MqttMessageEntity entity) {
		super.delete(entity);
		stats.incrementAndGet(MqttMessageDaoStat.MessagesDeleted);
	}

	@Override
	protected void setStoreStatementValues(MqttMessageEntity obj, PreparedStatement ps)
			throws SQLException {
		setInstantParameter(ps, 1, obj.getCreated() != null ? obj.getCreated() : Instant.now());
		setUpdateStatementValues(obj, ps, 1);
	}

	@Override
	protected void setUpdateStatementValues(MqttMessageEntity obj, PreparedStatement ps)
			throws SQLException {
		setUpdateStatementValues(obj, ps, 0);
		ps.setObject(6, obj.getId());
	}

	protected void setUpdateStatementValues(MqttMessageEntity obj, PreparedStatement ps, int offset)
			throws SQLException {
		ps.setString(1 + offset, obj.getDestination());
		ps.setString(2 + offset, obj.getTopic());
		ps.setBoolean(3 + offset, obj.isRetained());
		ps.setInt(4 + offset, obj.getQosLevel() != null ? obj.getQosLevel().getValue()
				: MqttQos.AtMostOnce.getValue());
		ps.setBytes(5 + offset, obj.getPayload());
	}

	/**
	 * A row mapper for {@link MqttMessageEntity} entities.
	 */
	public static final class MqttMessageEntityRowMapper implements RowMapper<MqttMessageEntity> {

		public static final RowMapper<MqttMessageEntity> INSTANCE = new MqttMessageEntityRowMapper();

		@Override
		public MqttMessageEntity mapRow(ResultSet rs, int rowNum) throws SQLException {
			Long id = rs.getLong(1);
			Instant created = getInstantColumn(rs, 2);
			String dest = rs.getString(3);
			String topic = rs.getString(4);
			boolean retained = rs.getBoolean(5);
			MqttQos qos = MqttQos.valueOf(rs.getInt(6));
			byte[] payload = rs.getBytes(7);

			return new BasicMqttMessageEntity(id, created, dest, topic, retained, qos, payload);
		}

	}

	@Transactional(readOnly = false, propagation = Propagation.REQUIRED)
	@Override
	public BatchResult batchProcess(BatchCallback<MqttMessageEntity> callback, BatchOptions options) {
		// wrap callback to track delete stats
		AtomicLong deleteCount = new AtomicLong(0);
		BatchResult result = super.batchProcess(new BatchCallback<MqttMessageEntity>() {

			@Override
			public BatchCallbackResult handle(MqttMessageEntity domainObject) {
				BatchCallbackResult result = callback.handle(domainObject);
				if ( result == BatchCallbackResult.DELETE ) {
					deleteCount.incrementAndGet();
				}
				return result;
			}

		}, options);
		getStats().addAndGet(MqttMessageDaoStat.MessagesDeleted, deleteCount.get());
		return result;
	}

	@Override
	protected String getBatchJdbcStatement(BatchOptions options) {
		Object destFilter = (options != null && options.getParameters() != null
				? options.getParameters().get(BATCH_OPTION_DESTINATION)
				: null);
		String resource;
		if ( options != null && options.isUpdatable() ) {
			if ( destFilter != null ) {
				resource = SqlResource.BatchUpdateForDestination.getResource();
			} else {
				resource = SqlResource.BatchUpdate.getResource();
			}
		} else if ( destFilter != null ) {
			resource = SqlResource.BatchForDestination.getResource();
		} else {
			resource = SqlResource.Batch.getResource();
		}
		return getSqlResource(resource);
	}

	@Override
	protected void prepareBatchStatement(BatchOptions options, Connection con,
			PreparedStatement queryStmt) throws SQLException {
		Object destFilter = (options != null && options.getParameters() != null
				? options.getParameters().get(BATCH_OPTION_DESTINATION)
				: null);
		if ( destFilter != null ) {
			queryStmt.setString(1, destFilter.toString());
		}
	}

	@Override
	protected MqttMessageEntity getBatchRowEntity(BatchOptions options, ResultSet resultSet,
			int rowCount) throws SQLException {
		return getRowMapper().mapRow(resultSet, rowCount);
	}

	@Override
	protected void updateBatchRowEntity(BatchOptions options, ResultSet resultSet, int rowCount,
			MqttMessageEntity entity) throws SQLException {
		resultSet.updateString(3, entity.getDestination());
		resultSet.updateString(4, entity.getTopic());
		resultSet.updateBoolean(5, entity.isRetained());
		resultSet.updateInt(6, entity.getQosLevel() != null ? entity.getQosLevel().getValue()
				: MqttQos.AtMostOnce.getValue());
		resultSet.updateBytes(7, entity.getPayload());
	}

	@Override
	public String getPingTestId() {
		return getSettingUid();
	}

	@Override
	public String getPingTestName() {
		return getDisplayName();
	}

	@Override
	public long getPingTestMaximumExecutionMilliseconds() {
		return 5000;
	}

	@Override
	public Result performPingTest() throws Exception {
		final long rowCount = rowCount();
		final int maxCount = getMaxCountPingFail();
		boolean ok = true;
		String msg = getMessageSource().getMessage("msg.messageCount", new Object[] { rowCount },
				Locale.getDefault());
		if ( maxCount > 0 && rowCount > maxCount ) {
			ok = false;
		}
		return new PingTestResult(ok, msg, Collections.singletonMap("count", rowCount));
	}

	/**
	 * Get the statistics object.
	 * 
	 * <p>
	 * This will have been initialized with {@link MqttMessageDaoStat} support.
	 * </p>
	 * 
	 * @return the statistics
	 */
	public MqttStats getStats() {
		return stats;
	}

	/**
	 * Get the maximum number of messages to store before failing ping tests.
	 * 
	 * @return the maximum count, or {@literal 0} to disable the test; defaults
	 *         to {@link #DEFAULT_MAX_COUNT_PING_FAIL}
	 * @since 2.1
	 */
	public int getMaxCountPingFail() {
		return maxCountPingFail;
	}

	/**
	 * Set the maximum number of messages to store before failing ping tests.
	 * 
	 * @param maxCountPingFail
	 *        the maximum count, or {@literal 0} to disable the test
	 * @since 2.1
	 */
	public void setMaxCountPingFail(int maxCountPingFail) {
		this.maxCountPingFail = maxCountPingFail;
	}

}
